package com.kancy.stream.rabbit.config;

import com.kancy.stream.rabbit.properties.ChannelProperties;
import com.kancy.stream.rabbit.properties.RabbitMessageProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.rabbit.listener.AbstractRabbitListenerEndpoint;
import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.util.StringUtils;

import java.util.Objects;

/**
 * RabbitMessageChannelListenerEndpoint
 *
 * @author kancy
 * @date 2020/3/8 0:52
 */
public class RabbitMessageChannelListenerEndpoint extends AbstractRabbitListenerEndpoint {
    private static Logger log = LoggerFactory.getLogger(RabbitMessageChannelListenerEndpoint.class);

    private final BeanFactory beanFactory;
    private final RabbitMessageProperties rabbitMessageProperties;
    private final ChannelProperties channelProperties;
    private final DynamicInterfaceBeanResolver dynamicInterfaceBeanResolver;

    private String channelKey;

    public RabbitMessageChannelListenerEndpoint(RabbitMessageProperties rabbitMessageProperties, String channelKey,
                                                BeanFactory beanFactory, DynamicInterfaceBeanResolver dynamicInterfaceBeanResolver) {
        this.beanFactory = beanFactory;
        this.channelKey = channelKey;
        this.channelProperties = rabbitMessageProperties.getChannel().get(channelKey);
        this.rabbitMessageProperties = rabbitMessageProperties;
        this.dynamicInterfaceBeanResolver = dynamicInterfaceBeanResolver;
        init();
        log.debug("Create rabbit listener endpoint [{}] on listener queue: {} success.", getId(), getQueueNames());
    }

    private void init() {
        setBeanFactory(beanFactory);
        setId(channelKey);
        setQueueNames(getQueueName());
        setConcurrency(String.valueOf(channelProperties.getConcurrency()));
        setAutoStartup(channelProperties.isAutoStartup());
        setExclusive(channelProperties.isExclusive());
        if (Objects.nonNull(channelProperties.getPriority())){
            setPriority(channelProperties.getPriority());
        }
        if (StringUtils.hasText(channelProperties.getGroup())){
            setGroup(channelProperties.getGroup());
        }
        if (StringUtils.hasText(channelProperties.getMessageConverter())){
            setMessageConverter(beanFactory.getBean(channelProperties.getMessageConverter(), MessageConverter.class));
        }

    }


    /**
     * 获取队列名称
     * @return
     */
    private String getQueueName() {
        String bindingKey = channelProperties.getBindingKey();

        RabbitMessageProperties.QueueProperties queueProperties = rabbitMessageProperties.getBindingQueue().get(bindingKey);
        if (Objects.isNull(queueProperties)){
            queueProperties = rabbitMessageProperties.getQueue().get(bindingKey);
        }

        if (Objects.nonNull(queueProperties)){
            if (StringUtils.hasText(queueProperties.getName())){
                return queueProperties.getName();
            }
            if (queueProperties instanceof RabbitMessageProperties.BindingQueueProperties){
                RabbitMessageProperties.BindingQueueProperties bindingQueueProperties = (RabbitMessageProperties.BindingQueueProperties) queueProperties;
                String exchangeName = bindingQueueProperties.getExchangeName();
                return String.format("%s.%s", exchangeName, bindingKey);
            }
            return String.format("%s.%s", rabbitMessageProperties.getProject(), bindingKey);
        }
        throw new IllegalArgumentException("bindingKey is error.");
    }

    /**
     * Create a {@link MessageListener} that is able to serve this endpoint for the
     * specified container.
     *
     * @param container the {@link MessageListenerContainer} to create a {@link MessageListener}.
     * @return a a {@link MessageListener} instance.
     */
    @Override
    protected MessageListener createMessageListener(MessageListenerContainer container) {
        return new RabbitMessageChannelListener(channelProperties, dynamicInterfaceBeanResolver);
    }

}
